{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "上次说了很多Linux下进程相关知识，这边不再复述，下面来说说Python的并发编程，如有错误欢迎提出～\n",
    "\n",
    "如果遇到听不懂的可以看上一次的文章：<a href=\"https://www.cnblogs.com/dotnetcrazy/p/9363810.html\" target=\"_blank\">https://www.cnblogs.com/dotnetcrazy/p/9363810.html</a>\n",
    "\n",
    "官方文档：<a href=\"https://docs.python.org/3/library/concurrency.html\" target=\"_blank\">https://docs.python.org/3/library/concurrency.html</a>\n",
    "\n",
    "## 1.进程篇\n",
    "\n",
    "官方文档：<a href=\"https://docs.python.org/3/library/multiprocessing.html\" target=\"_blank\">https://docs.python.org/3/library/multiprocessing.html</a>\n",
    "\n",
    "Code：<a href=\"https://github.com/lotapp/BaseCode/tree/master/python/5.concurrent/PythonProcess\" target=\"_blank\">https://github.com/lotapp/BaseCode/tree/master/python/5.concurrent/PythonProcess</a>\n",
    "\n",
    "### 1.1.进程（Process）\n",
    "\n",
    "Python的进程创建非常方便，看个案例：(这种方法通用，fork只适用于Linux系)\n",
    "```py\n",
    "import os\n",
    "# 注意一下，导入的是Process不是process（Class是大写开头）\n",
    "from multiprocessing import Process\n",
    "\n",
    "def test(name):\n",
    "    print(\"[子进程-%s]PID：%d，PPID：%d\" % (name, os.getpid(), os.getppid()))\n",
    "\n",
    "def main():\n",
    "    print(\"[父进程]PID：%d，PPID：%d\" % (os.getpid(), os.getppid()))\n",
    "    p = Process(target=test, args=(\"萌萌哒\", )) # 单个元素的元组表达别忘了(x,)\n",
    "    p.start()\n",
    "    p.join()  # 父进程回收子进程资源（内部调用了wait系列方法）\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    main()\n",
    "```\n",
    "运行结果：\n",
    "```\n",
    "[父进程]PID：25729，PPID：23434\n",
    "[子进程-萌萌哒]PID：25730，PPID：25729\n",
    "```\n",
    "\n",
    "创建子进程时，传入一个执行函数和参数，用start()方法来启动进程即可\n",
    "\n",
    "`join()`方法是父进程回收子进程的封装（主要是回收<a href=\"https://www.cnblogs.com/dotnetcrazy/p/9363810.html#2.2.僵尸进程和孤儿进程\" target=\"_blank\">僵尸子进程(点我)</a>）\n",
    "\n",
    "其他参数可以参考源码 or 文档，贴一下源码的`init`方法：\n",
    "\n",
    "`def __init__(self,group=None,target=None,name=None,args=(),kwargs={},*,daemon=None)`\n",
    "\n",
    "扩展：`name：为当前进程实例的别名`\n",
    "\n",
    "1. `p.is_alive()` 判断进程实例p是否还在执行\n",
    "2. `p.terminate()` 终止进程（发`SIGTERM`信号）\n",
    "\n",
    "上面的案例如果用OOP来实现就是这样：(如果不指定方法，默认调Run方法)\n",
    "```py\n",
    "import os\n",
    "from multiprocessing import Process\n",
    "\n",
    "class My_Process(Process):\n",
    "    # 重写了Proce类的Init方法\n",
    "    def __init__(self, name):\n",
    "        self.__name = name\n",
    "        Process.__init__(self)  # 调用父类方法\n",
    "\n",
    "    # 重写了Process类的run()方法\n",
    "    def run(self):\n",
    "        print(\"[子进程-%s]PID：%d，PPID：%d\" % (self.__name, os.getpid(),\n",
    "                                          os.getppid()))\n",
    "\n",
    "def main():\n",
    "    print(\"[父进程]PID：%d，PPID：%d\" % (os.getpid(), os.getppid()))\n",
    "    p = My_Process(\"萌萌哒\") # 如果不指定方法，默认调Run方法\n",
    "    p.start()\n",
    "    p.join()  # 父进程回收子进程资源（内部调用了wait系列方法）\n",
    "\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    main()\n",
    "```\n",
    "PS：`multiprocessing.Process`自行处理僵死进程，不用像`os.fork`那样自己建立信号处理程序、安装信号处理程序\n",
    "\n",
    "---\n",
    "\n",
    "### 1.1.源码拓展\n",
    "\n",
    "现在说说里面的一些门道（只想用的可以忽略）\n",
    "\n",
    "新版本的封装可能多层，这时候可以看看Python3.3.X系列（这个算是Python3早期版本了，很多代码都暴露出来，比较明了直观）\n",
    "\n",
    "multiprocessing.process.py\n",
    "```py\n",
    "# 3.4.x开始，Process有了一个BaseProcess\n",
    "# https://github.com/python/cpython/blob/3.7/Lib/multiprocessing/process.py\n",
    "# https://github.com/lotapp/cpython3/tree/master/Lib/multiprocessing/process.py\n",
    "def join(self, timeout=None):\n",
    "    '''一直等到子进程over'''\n",
    "    self._check_closed()\n",
    "    # 断言（False就触发异常，提示就是后面的内容\n",
    "    # 开发中用的比较多，部署的时候可以python3 -O xxx 去除所以断言\n",
    "    assert self._parent_pid == os.getpid(), \"只能 join 一个子进程\"\n",
    "    assert self._popen is not None, \"只能加入一个已启动的进程\"\n",
    "    res = self._popen.wait(timeout) # 本质就是用了我们之前讲的wait系列\n",
    "    if res is not None:\n",
    "        _children.discard(self) # 销毁子进程\n",
    "```\n",
    "multiprocessing.popen_fork.py\n",
    "```py\n",
    "# 3.4.x开始，在popen_fork文件中（以前是multiprocessing.forking.py）\n",
    "# https://github.com/python/cpython/blob/3.7/Lib/multiprocessing/popen_fork.py\n",
    "# https://github.com/lotapp/cpython3/tree/master/Lib/multiprocessing/popen_fork.py\n",
    "def wait(self, timeout=None):\n",
    "    if self.returncode is None:\n",
    "        # 设置超时的一系列处理\n",
    "        if timeout is not None:\n",
    "            from multiprocessing.connection import wait\n",
    "            if not wait([self.sentinel], timeout):\n",
    "                return None\n",
    "        # 核心操作\n",
    "        return self.poll(os.WNOHANG if timeout == 0.0 else 0)\n",
    "    return self.returncode\n",
    "\n",
    "# 回顾一下上次说的：os.WNOHANG - 如果没有子进程退出，则不阻塞waitpid()调用\n",
    "def poll(self, flag=os.WNOHANG):\n",
    "    if self.returncode is None:\n",
    "        try:\n",
    "            # 他的内部调用了waitpid\n",
    "            pid, sts = os.waitpid(self.pid, flag)\n",
    "        except OSError as e:\n",
    "            # 子进程尚未创建\n",
    "            # e.errno == errno.ECHILD == 10\n",
    "            return None\n",
    "        if pid == self.pid:\n",
    "            if os.WIFSIGNALED(sts):\n",
    "                self.returncode = -os.WTERMSIG(sts)\n",
    "            else:\n",
    "                assert os.WIFEXITED(sts), \"Status is {:n}\".format(sts)\n",
    "                self.returncode = os.WEXITSTATUS(sts)\n",
    "    return self.returncode\n",
    "```\n",
    "\n",
    "关于断言的简单说明：（别泛滥）\n",
    "\n",
    "如果条件为真，它什么都不做，反之它触发一个带可选错误信息的AssertionError\n",
    "\n",
    "```py\n",
    "def test(a, b):\n",
    "    assert b != 0, \"哥哥，分母不能为0啊\"\n",
    "    return a / b\n",
    "\n",
    "def main():\n",
    "    test(1, 0)\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    main()\n",
    "```\n",
    "结果：\n",
    "```\n",
    "Traceback (most recent call last):\n",
    "  File \"0.assert.py\", line 11, in <module>\n",
    "    main()\n",
    "  File \"0.assert.py\", line 7, in main\n",
    "    test(1, 0)\n",
    "  File \"0.assert.py\", line 2, in test\n",
    "    assert b != 0, \"哥哥，分母不能为0啊\"\n",
    "AssertionError: 哥哥，分母不能为0啊\n",
    "```\n",
    "运行的时候可以指定`-O参数`来忽略`assert`，eg：\n",
    "\n",
    "`python3 -O 0.assert.py `\n",
    "```\n",
    "Traceback (most recent call last):\n",
    "  File \"0.assert.py\", line 11, in <module>\n",
    "    main()\n",
    "  File \"0.assert.py\", line 7, in main\n",
    "    test(1, 0)\n",
    "  File \"0.assert.py\", line 3, in test\n",
    "    return a / b\n",
    "ZeroDivisionError: division by zero\n",
    "```\n",
    "\n",
    "---\n",
    "\n",
    "扩展：\n",
    "\n",
    "<a href=\"https://docs.python.org/3/library/unittest.html\" target=\"_blank\">https://docs.python.org/3/library/unittest.html</a>\n",
    "\n",
    "<a href=\"https://www.cnblogs.com/shangren/p/8038935.html\" target=\"_blank\">https://www.cnblogs.com/shangren/p/8038935.html</a>\n",
    "\n",
    "---\n",
    "\n",
    "### 1.2.进程池\n",
    "\n",
    "多个进程就不需要自己手动去管理了，有Pool来帮你完成，先看个案例：\n",
    "```py\n",
    "import os\n",
    "import time\n",
    "from multiprocessing import Pool  # 首字母大写\n",
    "\n",
    "def test(name):\n",
    "    print(\"[子进程-%s]PID=%d，PPID=%d\" % (name, os.getpid(), os.getppid()))\n",
    "    time.sleep(1)\n",
    "\n",
    "def main():\n",
    "    print(\"[父进程]PID=%d，PPID=%d\" % (os.getpid(), os.getppid()))\n",
    "    p = Pool(5) # 设置最多5个进程（不设置就默认为CPU核数）\n",
    "    for i in range(10):\n",
    "        # 异步执行\n",
    "        p.apply_async(test, args=(i, )) # 同步用apply（如非必要不建议用）\n",
    "    p.close() # 关闭池，不再加入新任务\n",
    "    p.join() # 等待所有子进程执行完毕回收资源（join可以指定超时时间，eg：`p.join(1)`）\n",
    "    print(\"over\")\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    main()\n",
    "```\n",
    "图示：（join可以指定超时时间，eg：`p.join(1)`）\n",
    "![1.进程池](https://images2018.cnblogs.com/blog/1127869/201808/1127869-20180805164841812-1549223582.gif)\n",
    "\n",
    "**调用`join()`之前必须先调用`close()`，调用`close()`之后就不能继续添加新的`Process`了**(<a href=\"#pool.join源码分析\" target=\"_blank\">下面会说为什么</a>)\n",
    "\n",
    "---\n",
    "\n",
    "### 1.3.源码拓展\n",
    "\n",
    "验证一下**Pool的默认大小是CPU的核数**，看源码：\n",
    "\n",
    "multiprocessing.pool.py\n",
    "```py\n",
    "# https://github.com/python/cpython/blob/3.7/Lib/multiprocessing/pool.py\n",
    "# https://github.com/lotapp/cpython3/tree/master/Lib/multiprocessing/pool.py\n",
    "class Pool(object):\n",
    "    def __init__(self, processes=指定的进程数,...):\n",
    "        if processes is None:\n",
    "            processes = os.cpu_count() or 1 # os.cpu_count() ~ CPU的核数\n",
    "```\n",
    "源码里面`apply_async`方法，是有回调函数（callback）的\n",
    "```py\n",
    "def apply_async(self,func,args=(),kwds={},callback=None,error_callback=None):\n",
    "    if self._state != RUN:\n",
    "        raise ValueError(\"Pool not running\")\n",
    "    result = ApplyResult(self._cache, callback, error_callback)\n",
    "    self._taskqueue.put(([(result._job, 0, func, args, kwds)], None))\n",
    "    return result\n",
    "```\n",
    "来看个例子：(和JQ很像)\n",
    "```py\n",
    "import os\n",
    "import time\n",
    "from multiprocessing import Pool  # 首字母大写\n",
    "\n",
    "def test(name):\n",
    "    print(\"[子进程%s]PID=%d，PPID=%d\" % (name, os.getpid(), os.getppid()))\n",
    "    time.sleep(1)\n",
    "    return name\n",
    "\n",
    "def error_test(name):\n",
    "    print(\"[子进程%s]PID=%d，PPID=%d\" % (name, os.getpid(), os.getppid()))\n",
    "    raise Exception(\"[子进程%s]啊，我挂了～\" % name)\n",
    "\n",
    "def callback(result):\n",
    "    \"\"\"成功之后的回调函数\"\"\"\n",
    "    print(\"[子进程%s]执行完毕\" % result)  # 没有返回值就为None\n",
    "\n",
    "def error_callback(msg):\n",
    "    \"\"\"错误之后的回调函数\"\"\"\n",
    "    print(msg)\n",
    "\n",
    "def main():\n",
    "    print(\"[父进程]PID=%d，PPID=%d\" % (os.getpid(), os.getppid()))\n",
    "    p = Pool()  # CPU默认核数\n",
    "    for i in range(5):\n",
    "        # 搞2个出错的看看\n",
    "        if i > 2:\n",
    "            p.apply_async(\n",
    "                error_test,\n",
    "                args=(i, ),\n",
    "                callback=callback,\n",
    "                error_callback=error_callback)  # 异步执行\n",
    "        else:\n",
    "            # 异步执行，成功后执行callback函数（有点像jq）\n",
    "            p.apply_async(test, args=(i, ), callback=callback)\n",
    "    p.close()  # 关闭池，不再加入新任务\n",
    "    p.join()  # 等待所有子进程执行完毕回收资源\n",
    "    print(\"over\")\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    main()\n",
    "```\n",
    "\n",
    "输出：\n",
    "```\n",
    "[父进程]PID=12348，PPID=10999\n",
    "[子进程0]PID=12349，PPID=12348\n",
    "[子进程2]PID=12351，PPID=12348\n",
    "[子进程1]PID=12350，PPID=12348\n",
    "[子进程3]PID=12352，PPID=12348\n",
    "[子进程4]PID=12352，PPID=12348\n",
    "[子进程3]啊，我挂了～\n",
    "[子进程4]啊，我挂了～\n",
    "[子进程0]执行完毕\n",
    "[子进程2]执行完毕\n",
    "[子进程1]执行完毕\n",
    "over\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "接着上面继续拓展，补充说说获取函数返回值。`上面是通过成功后的回调函数来获取返回值`，这次说说自带的方法：\n",
    "```py\n",
    "import time\n",
    "from multiprocessing import Pool, TimeoutError\n",
    "\n",
    "def test(x):\n",
    "    \"\"\"开平方\"\"\"\n",
    "    time.sleep(1)\n",
    "    return x * x\n",
    "\n",
    "def main():\n",
    "    pool = Pool()\n",
    "    task = pool.apply_async(test, (10, ))\n",
    "    print(task)\n",
    "    try:\n",
    "        print(task.get(timeout=1))\n",
    "    except TimeoutError as ex:\n",
    "        print(\"超时了～\", ex)\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    main()\n",
    "```\n",
    "输出：（`apply_async`返回一个`ApplyResult`类，里面有个get方法可以获取返回值）\n",
    "```\n",
    "<multiprocessing.pool.ApplyResult object at 0x7fbc354f50b8>\n",
    "超时了～\n",
    "```\n",
    "再举个例子，顺便把`Pool`里面的`map`和`imap`方法搞个案例（类比jq）\n",
    "```py\n",
    "import time\n",
    "from multiprocessing import Pool\n",
    "\n",
    "def test(x):\n",
    "    return x * x\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    with Pool(processes=4) as pool:\n",
    "        task = pool.apply_async(test, (10, ))\n",
    "        print(task.get(timeout=1))\n",
    "\n",
    "        obj_list = pool.map(test, range(10))\n",
    "        print(obj_list)\n",
    "        # 返回一个可迭代类的实例对象\n",
    "        obj_iter = pool.imap(test, range(10))\n",
    "        print(obj_iter)\n",
    "        next(obj_iter)\n",
    "        for i in obj_iter:\n",
    "            print(i, end=\" \")\n",
    "```\n",
    "输出：\n",
    "```\n",
    "100\n",
    "[0, 1, 4, 9, 16, 25, 36, 49, 64, 81]\n",
    "<multiprocessing.pool.IMapIterator object at 0x7ff7f9734198>\n",
    "1 4 9 16 25 36 49 64 81 \n",
    "```\n",
    "微微看一眼源码：(基础忘了可以查看==> <a href=\"https://www.cnblogs.com/dotnetcrazy/p/9278573.html\" target=\"_blank\">点我</a> )\n",
    "\n",
    "```py\n",
    "class IMapIterator(object):\n",
    "    def __init__(self, cache):\n",
    "        self._cond = threading.Condition(threading.Lock())\n",
    "        self._job = next(job_counter)\n",
    "        self._cache = cache\n",
    "        self._items = collections.deque()\n",
    "        self._index = 0\n",
    "        self._length = None\n",
    "        self._unsorted = {}\n",
    "        cache[self._job] = self\n",
    "    \n",
    "    def __iter__(self):\n",
    "        return self # 返回一个迭代器\n",
    "\n",
    "    # 实现next方法\n",
    "    def next(self, timeout=None):\n",
    "        with self._cond:\n",
    "            try:\n",
    "                item = self._items.popleft()\n",
    "            except IndexError:\n",
    "                if self._index == self._length:\n",
    "                    raise StopIteration from None\n",
    "                self._cond.wait(timeout)\n",
    "                try:\n",
    "                    item = self._items.popleft()\n",
    "                except IndexError:\n",
    "                    if self._index == self._length:\n",
    "                        raise StopIteration from None\n",
    "                    raise TimeoutError from None\n",
    "\n",
    "        success, value = item\n",
    "        if success:\n",
    "            return value\n",
    "        raise value\n",
    "......\n",
    "```\n",
    "\n",
    "扩展：优雅杀死子进程的探讨 <a href=\"https://segmentfault.com/q/1010000005077517\" target=\"_blank\">https://segmentfault.com/q/1010000005077517</a>\n",
    "\n",
    "---\n",
    "\n",
    "### 1.4.拓展之subprocess\n",
    "\n",
    "官方文档：<a href=\"https://docs.python.org/3/library/subprocess.html\" target=\"_blank\">https://docs.python.org/3/library/subprocess.html</a>\n",
    "\n",
    "还记得之前李代桃僵的`execlxxx`系列吗？\n",
    "\n",
    "这不，`subprocess`就是它的一层封装，当然了要强大的多，先看个例子：（以`os.execlp`的例子为引）\n",
    "```py\n",
    "import subprocess\n",
    "\n",
    "def main():\n",
    "    # os.execlp(\"ls\", \"ls\", \"-al\")  # 执行Path环境变量可以搜索到的命令\n",
    "    result = subprocess.run([\"ls\", \"-al\"])\n",
    "    print(result)\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    main()\n",
    "```\n",
    "输出\n",
    "```\n",
    "总用量 44\n",
    "drwxrwxr-x 2 dnt dnt 4096 8月   7 17:32 .\n",
    "drwxrwxr-x 4 dnt dnt 4096 8月   6 08:01 ..\n",
    "-rw-rw-r-- 1 dnt dnt  151 8月   3 10:49 0.assert.py\n",
    "-rw-rw-r-- 1 dnt dnt  723 8月   5 18:00 1.process2.py\n",
    "-rw-rw-r-- 1 dnt dnt  501 8月   3 10:20 1.process.py\n",
    "-rw-rw-r-- 1 dnt dnt 1286 8月   6 08:16 2.pool1.py\n",
    "-rw-rw-r-- 1 dnt dnt  340 8月   7 16:38 2.pool2.py\n",
    "-rw-rw-r-- 1 dnt dnt  481 8月   7 16:50 2.pool3.py\n",
    "-rw-rw-r-- 1 dnt dnt  652 8月   5 17:01 2.pool.py\n",
    "-rw-rw-r-- 1 dnt dnt  191 8月   7 17:33 3.subprocess.py\n",
    "CompletedProcess(args=['ls', '-al'], returncode=0)\n",
    "```\n",
    "#### 文档\n",
    "\n",
    "现在看下官方的文档描述来理解一下：\n",
    "```py\n",
    "r\"\"\"\n",
    "具有可访问I / O流的子进程\n",
    "Subprocesses with accessible I/O streams\n",
    "\n",
    "此模块允许您生成进程，连接到它们输入/输出/错误管道，并获取其返回代码。\n",
    "This module allows you to spawn processes, connect to their\n",
    "input/output/error pipes, and obtain their return codes.\n",
    "\n",
    "完整文档可以查看：https://docs.python.org/3/library/subprocess.html\n",
    "For a complete description of this module see the Python documentation.\n",
    "\n",
    "Main API\n",
    "========\n",
    "run(...): 运行命令，等待它完成，然后返回`CompletedProcess`实例。\n",
    "Runs a command, waits for it to complete, \n",
    "then returns a CompletedProcess instance.\n",
    "\n",
    "Popen(...): 用于在新进程中灵活执行命令的类\n",
    "A class for flexibly executing a command in a new process\n",
    "\n",
    "Constants（常量）\n",
    "---------\n",
    "DEVNULL: 特殊值，表示应该使用`os.devnull`\n",
    "Special value that indicates that os.devnull should be used\n",
    "\n",
    "PIPE:    表示应创建`PIPE`管道的特殊值\n",
    "Special value that indicates a pipe should be created\n",
    "\n",
    "STDOUT:  特殊值，表示`stderr`应该转到`stdout`\n",
    "Special value that indicates that stderr should go to stdout\n",
    "\n",
    "Older API（尽量不用，说不定以后就淘汰了）\n",
    "=========\n",
    "call(...): 运行命令，等待它完成，然后返回返回码。\n",
    "Runs a command, waits for it to complete, then returns the return code.\n",
    "\n",
    "check_call(...): Same as call() but raises CalledProcessError()\n",
    "    if return code is not 0（返回值不是0就引发异常）\n",
    "    \n",
    "check_output(...): 与check_call（）相同,但返回`stdout`的内容,而不是返回代码\n",
    "Same as check_call but returns the contents of stdout instead of a return code\n",
    "\n",
    "getoutput(...): 在shell中运行命令，等待它完成，然后返回输出\n",
    "Runs a command in the shell, waits for it to complete,then returns the output\n",
    "\n",
    "getstatusoutput(...): 在shell中运行命令，等待它完成，然后返回一个（exitcode，output）元组\n",
    "Runs a command in the shell, waits for it to complete,\n",
    "then returns a (exitcode, output) tuple\n",
    "\"\"\"\n",
    "```\n",
    "其实看看源码很有意思：（内部其实就是调用的`os.popen`【进程先导篇讲进程守护的时候用过】）\n",
    "```py\n",
    "def run(*popenargs, input=None, capture_output=False,\n",
    "        timeout=None, check=False, **kwargs):\n",
    "    \n",
    "    if input is not None:\n",
    "        if 'stdin' in kwargs:\n",
    "            raise ValueError('stdin和输入参数可能都不会被使用。')\n",
    "        kwargs['stdin'] = PIPE\n",
    "\n",
    "    if capture_output:\n",
    "        if ('stdout' in kwargs) or ('stderr' in kwargs):\n",
    "            raise ValueError('不能和capture_outpu一起使用stdout 或 stderr')\n",
    "        kwargs['stdout'] = PIPE\n",
    "        kwargs['stderr'] = PIPE\n",
    "\n",
    "    with Popen(*popenargs, **kwargs) as process:\n",
    "        try:\n",
    "            stdout, stderr = process.communicate(input, timeout=timeout)\n",
    "        except TimeoutExpired:\n",
    "            process.kill()\n",
    "            stdout, stderr = process.communicate()\n",
    "            raise TimeoutExpired(\n",
    "                process.args, timeout, output=stdout, stderr=stderr)\n",
    "        except:  # 包括KeyboardInterrupt的通信处理。\n",
    "            process.kill()\n",
    "            # 不用使用process.wait（），.__ exit__为我们做了这件事。\n",
    "            raise\n",
    "        retcode = process.poll()\n",
    "        if check and retcode:\n",
    "            raise CalledProcessError(\n",
    "                retcode, process.args, output=stdout, stderr=stderr)\n",
    "    return CompletedProcess(process.args, retcode, stdout, stderr)\n",
    "```\n",
    "返回值类型：`CompletedProcess`\n",
    "```py\n",
    "# https://github.com/lotapp/cpython3/blob/master/Lib/subprocess.py\n",
    "class CompletedProcess(object):\n",
    "    def __init__(self, args, returncode, stdout=None, stderr=None):\n",
    "        self.args = args\n",
    "        self.returncode = returncode\n",
    "        self.stdout = stdout\n",
    "        self.stderr = stderr\n",
    "\n",
    "    def __repr__(self):\n",
    "    \"\"\"对象按指定的格式显示\"\"\"\n",
    "        args = [\n",
    "            'args={!r}'.format(self.args),\n",
    "            'returncode={!r}'.format(self.returncode)\n",
    "        ]\n",
    "        if self.stdout is not None:\n",
    "            args.append('stdout={!r}'.format(self.stdout))\n",
    "        if self.stderr is not None:\n",
    "            args.append('stderr={!r}'.format(self.stderr))\n",
    "        return \"{}({})\".format(type(self).__name__, ', '.join(args))\n",
    "\n",
    "    def check_returncode(self):\n",
    "        \"\"\"如果退出代码非零，则引发CalledProcessError\"\"\"\n",
    "        if self.returncode:\n",
    "            raise CalledProcessError(self.returncode, self.args, self.stdout,\n",
    "                                     self.stderr)\n",
    "```\n",
    "\n",
    "#### 简单demo\n",
    "\n",
    "再来个案例体会一下方便之处：\n",
    "```py\n",
    "import subprocess\n",
    "\n",
    "def main():\n",
    "    result = subprocess.run([\"ping\", \"www.baidu.com\"])\n",
    "    print(result.stdout)\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    main()\n",
    "```\n",
    "图示：\n",
    "![2.subprocess.gif](https://images2018.cnblogs.com/blog/1127869/201808/1127869-20180807184228333-373477332.gif)\n",
    "\n",
    "#### 交互demo\n",
    "\n",
    "再来个强大的案例（交互的程序都可以，比如 **`ftp`，`nslookup`** 等等）：`popen1.communicate`\n",
    "```py\n",
    "import subprocess\n",
    "\n",
    "def main():\n",
    "    process = subprocess.Popen(\n",
    "        [\"ipython3\"],\n",
    "        stdin=subprocess.PIPE,\n",
    "        stdout=subprocess.PIPE,\n",
    "        stderr=subprocess.PIPE)\n",
    "    try:\n",
    "        # 对pstree进行交互\n",
    "        out, err = process.communicate(input=b'print(\"hello\")', timeout=3)\n",
    "        print(\"Out:%s\\nErr:%s\" % (out.decode(), err.decode()))\n",
    "    except TimeoutError:\n",
    "        # 如果超时到期，则子进程不会被终止，需要自己处理一下\n",
    "        process.kill()\n",
    "        out, err = process.communicate()\n",
    "        print(\"Out:%s\\nErr:%s\" % (out.decode(), err.decode()))\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    main()\n",
    "```\n",
    "输出：\n",
    "```\n",
    "IPython 6.4.0 -- An enhanced Interactive Python. Type '?' for help.\n",
    "\n",
    "In [1]: hello\n",
    "\n",
    "In [2]: Do you really want to exit ([y]/n)?\n",
    "\n",
    "Err:\n",
    "```\n",
    "注意点：如果超时到期，则子进程不会被终止，需要自己处理一下（官方提醒）\n",
    "\n",
    "\n",
    "#### 通信demo\n",
    "\n",
    "这个等会说进程间通信还会说，所以简单举个例子，老规矩拿`ps aux | grep bash`说事：\n",
    "```py\n",
    "import subprocess\n",
    "\n",
    "\n",
    "def main():\n",
    "    # ps aux | grep bash\n",
    "    # 进程1获取结果\n",
    "    p1 = subprocess.Popen([\"ps\", \"-aux\"], stdout=subprocess.PIPE)\n",
    "    # 得到进程1的结果再进行筛选\n",
    "    p2 = subprocess.Popen([\"grep\", \"bash\"], stdin=p1.stdout, stdout=subprocess.PIPE)\n",
    "    # 关闭写段（结果已经获取到进程2中了，防止干扰显示）\n",
    "    p1.stdout.close()\n",
    "    # 与流程交互：将数据发送到stdin并关闭它。\n",
    "    msg_tuple = p2.communicate()\n",
    "    # 输出结果\n",
    "    print(msg_tuple[0].decode())\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    main()\n",
    "```\n",
    "输出：（以前案例：<a href=\"https://www.cnblogs.com/dotnetcrazy/p/9363810.html#2.4.3.进程间通信～PIPE匿名管道（常用）\" target=\"_blank\">进程间通信～PIPE匿名管道</a>）\n",
    "```\n",
    "dnt       2470  0.0  0.1  24612  5236 pts/0    Ss   06:01   0:00 bash\n",
    "dnt       2512  0.0  0.1  24744  5760 pts/1    Ss   06:02   0:00 bash\n",
    "dnt      20784  0.0  0.1  24692  5588 pts/2    Ss+  06:21   0:00 /bin/bash\n",
    "dnt      22377  0.0  0.0  16180  1052 pts/1    S+   06:30   0:00 grep bash\n",
    "```\n",
    "\n",
    "其他扩展可以看看这篇文章：<a href=\"http://www.cnblogs.com/Security-Darren/p/4733368.html\" target=\"_blank\">subprocess与Popen()</a>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 1.5.进程间通信～PIPE管道通信\n",
    "\n",
    "这个比较有意思，看个案例：\n",
    "```py\n",
    "from multiprocessing import Process, Pipe\n",
    "\n",
    "def test(w):\n",
    "    w.send(\"[子进程]老爸，老妈回来记得喊我一下～\")\n",
    "    msg = w.recv()\n",
    "    print(msg)\n",
    "\n",
    "def main():\n",
    "    r, w = Pipe()\n",
    "    p1 = Process(target=test, args=(w, ))\n",
    "    p1.start()\n",
    "    msg = r.recv()\n",
    "    print(msg)\n",
    "    r.send(\"[父进程]滚犊子，赶紧写作业，不然我得跪方便面！\")\n",
    "    p1.join()\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    main()\n",
    "```\n",
    "结果：\n",
    "```\n",
    "老爸，老妈回来记得喊我一下～\n",
    "滚犊子，赶紧写作业，不然我得跪方便面！\n",
    "```\n",
    "#### multiprocessing.Pipe源码分析\n",
    "\n",
    "按照道理应该子进程自己写完自己读了，和上次讲得不一样啊？不急，先看看源码：\n",
    "```py\n",
    "# https://github.com/lotapp/cpython3/blob/master/Lib/multiprocessing/context.py\n",
    "def Pipe(self, duplex=True):\n",
    "    '''返回由管道连接的两个连接对象'''\n",
    "    from .connection import Pipe\n",
    "    return Pipe(duplex)\n",
    "```\n",
    "看看`connection.Pipe`方法的定义部分，是不是双向通信就看你是否设置`duplex=True`\n",
    "```py\n",
    "# https://github.com/lotapp/cpython3/blob/master/Lib/multiprocessing/connection.py\n",
    "if sys.platform != 'win32':\n",
    "    def Pipe(duplex=True):\n",
    "        '''返回管道两端的一对连接对象'''\n",
    "        if duplex:\n",
    "            # 双工内部其实是socket系列（下次讲）\n",
    "            s1, s2 = socket.socketpair()\n",
    "            s1.setblocking(True)\n",
    "            s2.setblocking(True)\n",
    "            c1 = Connection(s1.detach())\n",
    "            c2 = Connection(s2.detach())\n",
    "        else:\n",
    "            # 这部分就是我们上次讲的pipe管道\n",
    "            fd1, fd2 = os.pipe()\n",
    "            c1 = Connection(fd1, writable=False)\n",
    "            c2 = Connection(fd2, readable=False)\n",
    "        return c1, c2\n",
    "else: \n",
    "    def Pipe(duplex=True):\n",
    "        # win平台的一系列处理\n",
    "        ......\n",
    "        c1 = PipeConnection(h1, writable=duplex)\n",
    "        c2 = PipeConnection(h2, readable=duplex)\n",
    "        return c1, c2\n",
    "```\n",
    "通过源码知道了，原来双工是通过socket搞的啊～\n",
    "\n",
    "再看个和原来一样效果的案例：（不用关来关去的了，方便！）\n",
    "```py\n",
    "from multiprocessing import Process, Pipe\n",
    "\n",
    "def test(w):\n",
    "    # 只能写\n",
    "    w.send(\"[子进程]老爸，咱们完了，老妈一直在门口～\")\n",
    "\n",
    "def main():\n",
    "    r, w = Pipe(duplex=False)\n",
    "    p1 = Process(target=test, args=(w, ))\n",
    "    p1.start() # 你把这个放在join前面就直接死锁了\n",
    "    msg = r.recv() # 只能读\n",
    "    print(msg)\n",
    "    p1.join()\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    main()\n",
    "```\n",
    "输出：（可以思考下为什么`start换个位置就死锁`，提示：`阻塞读写`）\n",
    "```\n",
    "[子进程]老爸，咱们完了，老妈一直在门口～\n",
    "```\n",
    "\n",
    "再举个`Pool`的例子，咱们就进入今天的重点了：\n",
    "```py\n",
    "from multiprocessing import Pipe, Pool\n",
    "\n",
    "def proc_test1(conn):\n",
    "    conn.send(\"[小明]小张，今天哥们要见一女孩，你陪我呗，我24h等你回复哦～\")\n",
    "    msg = conn.recv()\n",
    "    print(msg)\n",
    "\n",
    "def proc_test2(conn):\n",
    "    msg = conn.recv()\n",
    "    print(msg)\n",
    "    conn.send(\"[小张]不去，万一被我帅气的外表迷倒就坑了～\")\n",
    "\n",
    "def main():\n",
    "    conn1, conn2 = Pipe()\n",
    "    p = Pool()\n",
    "    p.apply_async(proc_test1, (conn1, ))\n",
    "    p.apply_async(proc_test2, (conn2, ))\n",
    "    p.close()  # 关闭池，不再接收新任务\n",
    "    p.join()  # 等待回收，必须先关才能join，不然会异常\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    main()\n",
    "```\n",
    "输出：\n",
    "```py\n",
    "[小明]小张，今天哥们要见一女孩，你陪我呗，我24h等你回复哦～\n",
    "[小张]不去，万一被我帅气的外表迷倒就坑了～\n",
    "```\n",
    "#### pool.join源码分析\n",
    "\n",
    "看看源码就理解了：**看看Pool的join是啥情况？看源码：**\n",
    "```py\n",
    "# https://github.com/python/cpython/blob/3.7/Lib/multiprocessing/pool.py\n",
    "# https://github.com/lotapp/cpython3/blob/master/Lib/multiprocessing/pool.py\n",
    "def join(self):\n",
    "    util.debug('joining pool')\n",
    "    if self._state == RUN:\n",
    "        # 没关闭就join，这边就会抛出一个异常\n",
    "        raise ValueError(\"Pool is still running\")\n",
    "    elif self._state not in (CLOSE, TERMINATE):\n",
    "        raise ValueError(\"In unknown state\")\n",
    "    self._worker_handler.join()\n",
    "    self._task_handler.join()\n",
    "    self._result_handler.join()\n",
    "    for p in self._pool:\n",
    "        p.join() # 循环join回收\n",
    "```\n",
    "在pool的`__init__`的方法中，这几个属性：\n",
    "```py\n",
    "self._processes = processes # 指定的进程数\n",
    "self._pool = [] # 列表\n",
    "self._repopulate_pool() # 给列表append内容的方法\n",
    "```\n",
    "将池进程的数量增加到指定的数量，join的时候会使用这个列表\n",
    "```py\n",
    "def _repopulate_pool(self):\n",
    "    # 指定进程数-当前进程数，差几个补几个\n",
    "    for i in range(self._processes - len(self._pool)):\n",
    "        w = self.Process(target=worker,\n",
    "                         args=(self._inqueue, self._outqueue,\n",
    "                               self._initializer,\n",
    "                               self._initargs, self._maxtasksperchild,\n",
    "                               self._wrap_exception)\n",
    "                        )\n",
    "        self._pool.append(w) # 重点来了\n",
    "        w.name = w.name.replace('Process', 'PoolWorker')\n",
    "        w.daemon = True # pool退出后，通过pool创建的进程都会退出\n",
    "        w.start()\n",
    "        util.debug('added worker')\n",
    "```\n",
    "注意：**池的方法只能由创建它的进程使用**\n",
    "\n",
    "---\n",
    "\n",
    "### 1.5.进程间通信～Queue管道通信（常用）\n",
    "\n",
    "一步步的设局，从底层的的`pipe()`->`os.pipe`->`PIPE`，现在终于到`Queue`了，心酸啊，明知道上面两个项目\n",
    "\n",
    "里面基本上不会用，但为了你们能看懂源码，说了这么久`%>_<%`其实以后当我们从`Queue`说到`MQ`和`RPC`之后，现在\n",
    "\n",
    "讲得这些进程间通信(`IPC`)也基本上不会用了，但本质你得清楚，我尽量多分析点源码，这样你们以后看开源项目压力会很小\n",
    "\n",
    "欢迎批评指正～\n",
    "\n",
    "#### 引入案例\n",
    "\n",
    "```py\n",
    "from multiprocessing import Process, Queue\n",
    "\n",
    "def test(q):\n",
    "    q.put(\"[子进程]老爸，我出去嗨了\")\n",
    "    print(q.get())\n",
    "\n",
    "def main():\n",
    "    q = Queue()\n",
    "    p = Process(target=test, args=(q, ))\n",
    "    p.start()\n",
    "    msg = q.get()\n",
    "    print(msg)\n",
    "    q.put(\"[父进程]去吧比卡丘～\")\n",
    "    p.join()\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    main()\n",
    "```\n",
    "输出：（`get`和`put`默认是阻塞等待的）\n",
    "```\n",
    "[子进程]老爸，我出去嗨了\n",
    "[父进程]去吧比卡丘～\n",
    "```\n",
    "\n",
    "#### 源码拓展\n",
    "\n",
    "先看看`Queue`的初始化方法：（不指定大小就是最大队列数）\n",
    "```py\n",
    "# 队列类型，使用PIPE，缓存，线程\n",
    "class Queue(object):\n",
    "    # ctx = multiprocessing.get_context(\"xxx\")\n",
    "    # 上下文总共3种：spawn、fork、forkserver（扩展部分会提一下）\n",
    "    def __init__(self, maxsize=0, *, ctx):\n",
    "        # 默认使用最大容量\n",
    "        if maxsize <= 0:\n",
    "            from .synchronize import SEM_VALUE_MAX as maxsize\n",
    "        self._maxsize = maxsize  # 指定队列大小\n",
    "        # 创建了一个PIPE匿名管道（单向）\n",
    "        self._reader, self._writer = connection.Pipe(duplex=False)\n",
    "        # `multiprocessing/synchronize.py > Lock`\n",
    "        self._rlock = ctx.Lock()  # 进程锁（读）【非递归】\n",
    "        self._opid = os.getpid()  # 获取PID\n",
    "        if sys.platform == 'win32':\n",
    "            self._wlock = None\n",
    "        else:\n",
    "            self._wlock = ctx.Lock()  # 进程锁（写）【非递归】\n",
    "        # Semaphore信号量通常用于保护容量有限的资源\n",
    "        # 控制信号量,超了就异常\n",
    "        self._sem = ctx.BoundedSemaphore(maxsize)\n",
    "        # 不忽略PIPE管道破裂的错误\n",
    "        self._ignore_epipe = False \n",
    "        # 线程相关操作\n",
    "        self._after_fork()\n",
    "        # 向`_afterfork_registry`字典中注册\n",
    "        if sys.platform != 'win32':\n",
    "            register_after_fork(self, Queue._after_fork)\n",
    "```\n",
    "关于`get`和`put`是阻塞的问题，看下源码探探究竟：\n",
    "\n",
    "`q.get()`：收消息\n",
    "```py\n",
    "def get(self, block=True, timeout=None):\n",
    "    # 默认情况是阻塞（lock加锁）\n",
    "    if block and timeout is None:\n",
    "        with self._rlock:\n",
    "            res = self._recv_bytes()\n",
    "        self._sem.release()  # 信号量+1\n",
    "    else:\n",
    "        if block:\n",
    "            deadline = time.monotonic() + timeout\n",
    "        # 超时抛异常\n",
    "        if not self._rlock.acquire(block, timeout):\n",
    "            raise Empty\n",
    "        try:\n",
    "            if block:\n",
    "                timeout = deadline - time.monotonic()\n",
    "                # 不管有没有内容都去读，超时就抛异常\n",
    "                if not self._poll(timeout):\n",
    "                    raise Empty\n",
    "            elif not self._poll():\n",
    "                raise Empty\n",
    "            # 接收字节数据作为字节对象\n",
    "            res = self._recv_bytes()\n",
    "            self._sem.release()  # 信号量+1\n",
    "        finally:\n",
    "            # 释放锁\n",
    "            self._rlock.release()\n",
    "    # 释放锁后，重新序列化数据\n",
    "    return _ForkingPickler.loads(res)\n",
    "```\n",
    "`queue.put()`:发消息\n",
    "```py\n",
    "def put(self, obj, block=True, timeout=None):\n",
    "        # 如果Queue已经关闭就抛异常\n",
    "        assert not self._closed, \"Queue {0!r} has been closed\".format(self)\n",
    "        # 记录信号量的锁\n",
    "        if not self._sem.acquire(block, timeout):\n",
    "            raise Full  # 超过数量，抛个异常\n",
    "        # 条件变量允许一个或多个线程等待，直到另一个线程通知它们\n",
    "        with self._notempty:\n",
    "            if self._thread is None:\n",
    "                self._start_thread()\n",
    "            self._buffer.append(obj)\n",
    "            self._notempty.notify()\n",
    "```\n",
    "\n",
    "非阻塞`get_nowait`和`put_nowait`本质其实也是调用了`get`和`put`方法：\n",
    "```py\n",
    "def get_nowait(self):\n",
    "    return self.get(False)\n",
    "\n",
    "def put_nowait(self, obj):\n",
    "    return self.put(obj, False)\n",
    "```\n",
    "\n",
    "#### 进程间通信1\n",
    "\n",
    "说这么多不如来个例子看看：\n",
    "```py\n",
    "from multiprocessing import Queue\n",
    "\n",
    "def main():\n",
    "    q = Queue(3)  # 只能 put 3条消息\n",
    "    q.put([1, 2, 3, 4])  # put一个List类型的消息\n",
    "    q.put({\"a\": 1, \"b\": 2})  # put一个Dict类型的消息\n",
    "    q.put({1, 2, 3, 4})  # put一个Set类型的消息\n",
    "\n",
    "    try:\n",
    "        # 不加timeout，就一直阻塞，等消息队列有空位才能发出去\n",
    "        q.put(\"再加条消息呗\", timeout=2)\n",
    "    # Full(Exception)是空实现，你可以直接用Exception\n",
    "    except Exception:\n",
    "        print(\"消息队列已满，队列数%s，当前存在%s条消息\" % (q._maxsize, q.qsize()))\n",
    "\n",
    "    try:\n",
    "        # 非阻塞，不能put就抛异常\n",
    "        q.put_nowait(\"再加条消息呗\")  # 相当于q.put(obj,False)\n",
    "    except Exception:\n",
    "        print(\"消息队列已满，队列数%s，当前存在%s条消息\" % (q._maxsize, q.qsize()))\n",
    "\n",
    "    while not q.empty():\n",
    "        print(\"队列数：%s，当前存在%s条消息 内容%s\" % (q._maxsize, q.qsize(), q.get_nowait()))\n",
    "\n",
    "    print(\"队列数：%s，当前存在：%s条消息\" % (q._maxsize, q.qsize()))\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    main()\n",
    "```\n",
    "输出：\n",
    "```\n",
    "消息队列已满，队列数3，当前存在3条消息\n",
    "消息队列已满，队列数3，当前存在3条消息\n",
    "队列数：3，当前存在3条消息 内容[1, 2, 3, 4]\n",
    "队列数：3，当前存在2条消息 内容{'a': 1, 'b': 2}\n",
    "队列数：3，当前存在1条消息 内容{1, 2, 3, 4}\n",
    "队列数：3，当前存在：0条消息\n",
    "```\n",
    "补充说明一下：\n",
    "1. *`q._maxsize`* 队列数(尽量不用`_`开头的属性和方法）\n",
    "2. `q.qsize()`查看当前队列中存在几条消息\n",
    "3. `q.full()`查看是否满了\n",
    "4. `q.empty()`查看是否为空\n",
    "\n",
    "再看个简单点的子进程间通信：(铺垫demo)\n",
    "```py\n",
    "import os\n",
    "import time\n",
    "from multiprocessing import Process, Queue\n",
    "\n",
    "def pro_test1(q):\n",
    "    print(\"[子进程1]PPID=%d,PID=%d,GID=%d\"%(os.getppid(), os.getpid(), os.getgid()))\n",
    "    q.put(\"[子进程1]小明，今晚撸串不？\")\n",
    "\n",
    "    # 设置一个简版的重试机制（三次重试）\n",
    "    for i in range(3):\n",
    "        if not q.empty():\n",
    "            print(q.get())\n",
    "            break\n",
    "        else:\n",
    "            time.sleep((i + 1) * 2)  # 第一次1s，第二次4s，第三次6s\n",
    "\n",
    "def pro_test2(q):\n",
    "    print(\"[子进程2]PPID=%d,PID=%d,GID=%d\"%(os.getppid(), os.getpid(), os.getgid()))\n",
    "    print(q.get())\n",
    "    time.sleep(4)  # 模拟一下网络延迟\n",
    "    q.put(\"[子进程2]不去，我今天约了妹子\")\n",
    "\n",
    "def main():\n",
    "    queue = Queue()\n",
    "    p1 = Process(target=pro_test1, args=(queue, ))\n",
    "    p2 = Process(target=pro_test2, args=(queue, ))\n",
    "    p1.start()\n",
    "    p2.start()\n",
    "    p1.join()\n",
    "    p2.join()\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    main()\n",
    "```\n",
    "输出：（`time python3 5.queue2.py`）\n",
    "```\n",
    "[子进程1]PPID=15220,PID=15221,GID=1000\n",
    "[子进程2]PPID=15220,PID=15222,GID=1000\n",
    "[子进程1]小明，今晚撸串不？\n",
    "[子进程2]不去，我今天约了妹子\n",
    "\n",
    "real\t0m6.087s\n",
    "user\t0m0.053s\n",
    "sys\t0m0.035s\n",
    "```\n",
    "\n",
    "---\n",
    "\n",
    "#### 进程间通信2\n",
    "\n",
    "多进程基本上都是用`pool`，可用上面说的`Queue`方法怎么报错了？\n",
    "```py\n",
    "import os\n",
    "import time\n",
    "from multiprocessing import Pool, Queue\n",
    "\n",
    "def error_callback(msg):\n",
    "    print(msg)\n",
    "\n",
    "def pro_test1(q):\n",
    "    print(\"[子进程1]PPID=%d,PID=%d,GID=%d\" % (os.getppid(), os.getpid(),\n",
    "                                           os.getgid()))\n",
    "    q.put(\"[子进程1]小明，今晚撸串不？\")\n",
    "\n",
    "    # 设置一个简版的重试机制（三次重试）\n",
    "    for i in range(3):\n",
    "        if not q.empty():\n",
    "            print(q.get())\n",
    "            break\n",
    "        else:\n",
    "            time.sleep((i + 1) * 2)  # 第一次1s，第二次4s，第三次6s\n",
    "\n",
    "def pro_test2(q):\n",
    "    print(\"[子进程2]PPID=%d,PID=%d,GID=%d\" % (os.getppid(), os.getpid(),\n",
    "                                           os.getgid()))\n",
    "    print(q.get())\n",
    "    time.sleep(4)  # 模拟一下网络延迟\n",
    "    q.put(\"[子进程2]不去，我今天约了妹子\")\n",
    "\n",
    "def main():\n",
    "    print(\"[父进程]PPID=%d,PID=%d,GID=%d\" % (os.getppid(), os.getpid(),\n",
    "                                          os.getgid()))\n",
    "    queue = Queue()\n",
    "    p = Pool()\n",
    "    p.apply_async(pro_test1, args=(queue, ), error_callback=error_callback)\n",
    "    p.apply_async(pro_test2, args=(queue, ), error_callback=error_callback)\n",
    "    p.close()\n",
    "    p.join()\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    main()\n",
    "```\n",
    "输出：(无法将`multiprocessing.Queue`对象传递给`Pool`方法)\n",
    "```\n",
    "[父进程]PPID=4223,PID=32170,GID=1000\n",
    "Queue objects should only be shared between processes through inheritance\n",
    "Queue objects should only be shared between processes through inheritance\n",
    "\n",
    "real\t0m0.183s\n",
    "user\t0m0.083s\n",
    "sys\t0m0.012s\n",
    "```\n",
    "下面会详说，先看一下正确方式：（队列换了一下，其他都一样`Manager().Queue()`）\n",
    "```py\n",
    "import os\n",
    "import time\n",
    "from multiprocessing import Pool, Manager\n",
    "\n",
    "def error_callback(msg):\n",
    "    print(msg)\n",
    "\n",
    "def pro_test1(q):\n",
    "    print(\"[子进程1]PPID=%d,PID=%d,GID=%d\" % (os.getppid(), os.getpid(),\n",
    "                                           os.getgid()))\n",
    "    q.put(\"[子进程1]小明，今晚撸串不？\")\n",
    "\n",
    "    # 设置一个简版的重试机制（三次重试）\n",
    "    for i in range(3):\n",
    "        if not q.empty():\n",
    "            print(q.get())\n",
    "            break\n",
    "        else:\n",
    "            time.sleep((i + 1) * 2)  # 第一次1s，第二次4s，第三次6s\n",
    "\n",
    "def pro_test2(q):\n",
    "    print(\"[子进程2]PPID=%d,PID=%d,GID=%d\" % (os.getppid(), os.getpid(),\n",
    "                                           os.getgid()))\n",
    "    print(q.get())\n",
    "    time.sleep(4)  # 模拟一下网络延迟\n",
    "    q.put(\"[子进程2]不去，我今天约了妹子\")\n",
    "\n",
    "def main():\n",
    "    print(\"[父进程]PPID=%d,PID=%d,GID=%d\" % (os.getppid(), os.getpid(),\n",
    "                                          os.getgid()))\n",
    "    queue = Manager().Queue()\n",
    "    p = Pool()\n",
    "    p.apply_async(pro_test1, args=(queue, ), error_callback=error_callback)\n",
    "    p.apply_async(pro_test2, args=(queue, ), error_callback=error_callback)\n",
    "    p.close()\n",
    "    p.join()\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    main()\n",
    "```\n",
    "输出：\n",
    "```\n",
    "[父进程]PPID=4223,PID=31329,GID=1000\n",
    "[子进程1]PPID=31329,PID=31335,GID=1000\n",
    "[子进程2]PPID=31329,PID=31336,GID=1000\n",
    "[子进程1]小明，今晚撸串不？\n",
    "[子进程2]不去，我今天约了妹子\n",
    "\n",
    "real\t0m6.134s\n",
    "user\t0m0.133s\n",
    "sys\t0m0.035s\n",
    "```\n",
    "再抛个思考题：（Linux）\n",
    "```py\n",
    "import os\n",
    "import time\n",
    "from multiprocessing import Pool, Queue\n",
    "\n",
    "def error_callback(msg):\n",
    "    print(msg)\n",
    "\n",
    "q = Queue()\n",
    "\n",
    "def pro_test1():\n",
    "    global q\n",
    "    print(\"[子进程1]PPID=%d,PID=%d,GID=%d\" % (os.getppid(), os.getpid(),\n",
    "                                           os.getgid()))\n",
    "    q.put(\"[子进程1]小明，今晚撸串不？\")\n",
    "    # 设置一个简版的重试机制（三次重试）\n",
    "    for i in range(3):\n",
    "        if not q.empty():\n",
    "            print(q.get())\n",
    "            break\n",
    "        else:\n",
    "            time.sleep((i + 1) * 2)  # 第一次1s，第二次4s，第三次6s\n",
    "\n",
    "def pro_test2():\n",
    "    global q\n",
    "    print(\"[子进程2]PPID=%d,PID=%d,GID=%d\" % (os.getppid(), os.getpid(),\n",
    "                                           os.getgid()))\n",
    "    print(q.get())\n",
    "    time.sleep(4)  # 模拟一下网络延迟\n",
    "    q.put(\"[子进程2]不去，我今天约了妹子\")\n",
    "\n",
    "def main():\n",
    "    print(\"[父进程]PPID=%d,PID=%d,GID=%d\" % (os.getppid(), os.getpid(),\n",
    "                                          os.getgid()))\n",
    "    q = Queue()\n",
    "    p = Pool()\n",
    "    p.apply_async(pro_test1, error_callback=error_callback)\n",
    "    p.apply_async(pro_test2, error_callback=error_callback)\n",
    "    p.close()\n",
    "    p.join()\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    main()\n",
    "```\n",
    "输出：（为啥这样也可以【提示：`fork`】）\n",
    "```\n",
    "[父进程]PPID=12855,PID=16879,GID=1000\n",
    "[子进程1]PPID=16879,PID=16880,GID=1000\n",
    "[子进程2]PPID=16879,PID=16881,GID=1000\n",
    "[子进程1]小明，今晚撸串不？\n",
    "[子进程2]不去，我今天约了妹子\n",
    "\n",
    "real    0m6.120s\n",
    "user    0m0.105s\n",
    "sys     0m0.024s\n",
    "```\n",
    "\n",
    "### 进程拓展\n",
    "\n",
    "官方参考：<a href=\"https://docs.python.org/3/library/multiprocessing.html\" target=\"_blank\">https://docs.python.org/3/library/multiprocessing.html</a>\n",
    "\n",
    "#### 1.上下文系\n",
    "\n",
    "1. spawn：（Win默认，Linux下也可以用【>=3.4】）\n",
    "    1. 父进程启动一个新的python解释器进程。\n",
    "    2. 子进程只会继承运行进程对象run()方法所需的那些资源。\n",
    "    3. 不会继承父进程中不必要的文件描述符和句柄。\n",
    "    4. 与使用fork或forkserver相比，使用此方法启动进程相当慢。\n",
    "    5. 可在Unix和Windows上使用。Windows上的默认设置。\n",
    "2. fork:（Linux下默认）\n",
    "    1. 父进程用于os.fork()分叉Python解释器。\n",
    "    2. 子进程在开始时与父进程相同（这时候内部变量之类的还没有被修改）\n",
    "    3. 父进程的所有资源都由子进程继承（用到多线程的时候可能有些问题）\n",
    "    4. 仅适用于Unix。Unix上的默认值。\n",
    "3. forkserver：（常用）\n",
    "    1. 当程序启动并选择forkserver start方法时，将启动服务器进程。\n",
    "    2. 从那时起，每当需要一个新进程时，父进程就会连接到服务器并请求它分叉一个新进程。\n",
    "    3. fork服务器进程是单线程的，因此它可以安全使用os.fork()。没有不必要的资源被继承。\n",
    "    4. 可在Unix平台上使用，支持通过Unix管道传递文件描述符。\n",
    "\n",
    "这块官方文档很详细，贴下官方的2个案例：\n",
    "\n",
    "通过`multiprocessing.set_start_method(xxx)`来设置启动的上下文类型\n",
    "```py\n",
    "import multiprocessing as mp\n",
    "\n",
    "def foo(q):\n",
    "    q.put('hello')\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    mp.set_start_method('spawn') # 不要过多使用\n",
    "    q = mp.Queue()\n",
    "    p = mp.Process(target=foo, args=(q,))\n",
    "    p.start()\n",
    "    print(q.get())\n",
    "    p.join()\n",
    "```\n",
    "输出：（`set_start_method`不要过多使用）\n",
    "```\n",
    "hello\n",
    "\n",
    "real\t0m0.407s\n",
    "user\t0m0.134s\n",
    "sys\t    0m0.012s\n",
    "```\n",
    "如果你把设置启动上下文注释掉：（消耗的总时间少了很多）\n",
    "```\n",
    "real\t0m0.072s\n",
    "user\t0m0.057s\n",
    "sys\t    0m0.016s\n",
    "```\n",
    "\n",
    "也可以通过`multiprocessing.get_context(xxx)`获取指定类型的上下文\n",
    "```py\n",
    "import multiprocessing as mp\n",
    "\n",
    "def foo(q):\n",
    "    q.put('hello')\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    ctx = mp.get_context('spawn')\n",
    "    q = ctx.Queue()\n",
    "    p = ctx.Process(target=foo, args=(q,))\n",
    "    p.start()\n",
    "    print(q.get())\n",
    "    p.join()\n",
    "```\n",
    "输出：（`get_context`在Python源码里用的比较多，so=>也建议大家这么用）\n",
    "```\n",
    "hello\n",
    "\n",
    "real\t0m0.169s\n",
    "user\t0m0.146s\n",
    "sys\t0m0.024s\n",
    "```\n",
    "从结果来看，总耗时也少了很多\n",
    "\n",
    "---\n",
    "\n",
    "#### 2.日记系列\n",
    "\n",
    "说下日记相关的事情：\n",
    "\n",
    "先看下`multiprocessing`里面的日记记录：\n",
    "```py\n",
    "# https://github.com/lotapp/cpython3/blob/master/Lib/multiprocessing/context.py\n",
    "def log_to_stderr(self, level=None):\n",
    "    '''打开日志记录并添加一个打印到stderr的处理程序'''\n",
    "    from .util import log_to_stderr\n",
    "    return log_to_stderr(level)\n",
    "```\n",
    "更多`Loging`模块内容可以看官方文档：<a href=\"https://docs.python.org/3/library/logging.html\" target=\"_blank\">https://docs.python.org/3/library/logging.html</a>\n",
    "\n",
    "这个是内部代码，看看即可：\n",
    "```py\n",
    "# https://github.com/lotapp/cpython3/blob/master/Lib/multiprocessing/util.py\n",
    "def log_to_stderr(level=None):\n",
    "    '''打开日志记录并添加一个打印到stderr的处理程序'''\n",
    "    # 全局变量默认是False\n",
    "    global _log_to_stderr\n",
    "    import logging\n",
    "    \n",
    "    # 日记记录转换成文本\n",
    "    formatter = logging.Formatter(DEFAULT_LOGGING_FORMAT)\n",
    "    # 一个处理程序类，它将已适当格式化的日志记录写入流\n",
    "    handler = logging.StreamHandler()  # 此类不会关闭流，因为用到了sys.stdout|sys.stderr\n",
    "    # 设置格式：'[%(levelname)s/%(processName)s] %(message)s'\n",
    "    handler.setFormatter(formatter)\n",
    "\n",
    "    # 返回`multiprocessing`专用的记录器\n",
    "    logger = get_logger()\n",
    "    # 添加处理程序\n",
    "    logger.addHandler(handler)\n",
    "\n",
    "    if level:\n",
    "        # 设置日记级别\n",
    "        logger.setLevel(level)\n",
    "    # 现在log是输出到stderr的\n",
    "    _log_to_stderr = True\n",
    "    return _logger\n",
    "```\n",
    "`Logging`之前也有提过，可以看看：<a href=\"https://www.cnblogs.com/dotnetcrazy/p/9333792.html#2.装饰器传参的扩展（可传可不传）\" target=\"_blank\">https://www.cnblogs.com/dotnetcrazy/p/9333792.html#2.装饰器传参的扩展（可传可不传）</a>\n",
    "\n",
    "来个案例：\n",
    "```py\n",
    "import logging\n",
    "from multiprocessing import Process, log_to_stderr\n",
    "\n",
    "def test():\n",
    "    print(\"test\")\n",
    "\n",
    "def start_log():\n",
    "    # 把日记输出定向到sys.stderr中\n",
    "    logger = log_to_stderr()\n",
    "    # 设置日记记录级别\n",
    "    # 敏感程度：DEBUG、INFO、WARN、ERROR、CRITICAL\n",
    "    print(logging.WARN == logging.WARNING)  # 这两个是一样的\n",
    "    level = logging.INFO\n",
    "    logger.setLevel(level)  # 设置日记级别(一般都是WARN)\n",
    "\n",
    "    # 自定义输出\n",
    "    # def log(self, level, msg, *args, **kwargs):\n",
    "    logger.log(level, \"我是通用格式\")  # 通用，下面的内部也是调用的这个\n",
    "    logger.info(\"info 测试\")\n",
    "    logger.warning(\"warning 测试\")\n",
    "    logger.error(\"error 测试\")\n",
    "\n",
    "def main():\n",
    "    start_log()\n",
    "    # 做的操作都会被记录下来\n",
    "    p = Process(target=test)\n",
    "    p.start()\n",
    "    p.join()\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    main()\n",
    "```\n",
    "输出：\n",
    "```\n",
    "True\n",
    "[INFO/MainProcess] 我是通用格式\n",
    "[INFO/MainProcess] info 测试\n",
    "[WARNING/MainProcess] warning 测试\n",
    "[ERROR/MainProcess] error 测试\n",
    "[INFO/Process-1] child process calling self.run()\n",
    "test\n",
    "[INFO/Process-1] process shutting down\n",
    "[INFO/Process-1] process exiting with exitcode 0\n",
    "[INFO/MainProcess] process shutting down\n",
    "```\n",
    "\n",
    "---\n",
    "\n",
    "#### 3.进程5态\n",
    "\n",
    "之前忘记说了～现在快结尾了，补充一下进程5态：(来个草图)\n",
    "\n",
    "![3.进程5态.png](https://images2018.cnblogs.com/blog/1127869/201808/1127869-20180810162418968-770691664.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 1.6.进程间状态共享\n",
    "\n",
    "应该尽量避免进程间状态共享，但需求在那，所以还是得研究，官方推荐了两种方式：\n",
    "\n",
    "#### 1.共享内存（`Value` or `Array`）\n",
    "\n",
    "之前说过`Queue`：在`Process`之间使用没问题，用到`Pool`，就使用`Manager().xxx`，`Value`和`Array`，就不太一样了：\n",
    "\n",
    "看看源码：（Manager里面的Array和Process共享的Array不是一个概念，而且也没有同步机制）\n",
    "```py\n",
    "# https://github.com/lotapp/cpython3/blob/master/Lib/multiprocessing/managers.py\n",
    "class Value(object):\n",
    "    def __init__(self, typecode, value, lock=True):\n",
    "        self._typecode = typecode\n",
    "        self._value = value\n",
    "\n",
    "    def get(self):\n",
    "        return self._value\n",
    "\n",
    "    def set(self, value):\n",
    "        self._value = value\n",
    "\n",
    "    def __repr__(self):\n",
    "        return '%s(%r, %r)' % (type(self).__name__, self._typecode, self._value)\n",
    "\n",
    "    value = property(get, set) # 给value设置get和set方法（和value的属性装饰器一样效果）\n",
    "\n",
    "def Array(typecode, sequence, lock=True):\n",
    "    return array.array(typecode, sequence)\n",
    "```\n",
    "以`Process`为例看看怎么用：\n",
    "```py\n",
    "from multiprocessing import Process, Value, Array\n",
    "\n",
    "def proc_test1(value, array):\n",
    "    print(\"子进程1\", value.value)\n",
    "    array[0] = 10\n",
    "    print(\"子进程1\", array[:])\n",
    "\n",
    "def proc_test2(value, array):\n",
    "    print(\"子进程2\", value.value)\n",
    "    array[1] = 10\n",
    "    print(\"子进程2\", array[:])\n",
    "\n",
    "def main():\n",
    "    try:\n",
    "        value = Value(\"d\", 3.14)  # d 类型，相当于C里面的double\n",
    "        array = Array(\"i\", range(10))  # i 类型，相当于C里面的int\n",
    "        print(type(value))\n",
    "        print(type(array))\n",
    "\n",
    "        p1 = Process(target=proc_test1, args=(value, array))\n",
    "        p2 = Process(target=proc_test2, args=(value, array))\n",
    "        p1.start()\n",
    "        p2.start()\n",
    "        p1.join()\n",
    "        p2.join()\n",
    "\n",
    "        print(\"父进程\", value.value)  # 获取值\n",
    "        print(\"父进程\", array[:])  # 获取值\n",
    "    except Exception as ex:\n",
    "        print(ex)\n",
    "    else:\n",
    "        print(\"No Except\")\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    main()\n",
    "```\n",
    "输出：（`Value`和`Array`是`进程|线程`安全的）\n",
    "```py\n",
    "<class 'multiprocessing.sharedctypes.Synchronized'>\n",
    "<class 'multiprocessing.sharedctypes.SynchronizedArray'>\n",
    "子进程1 3.14\n",
    "子进程1 [10, 1, 2, 3, 4, 5, 6, 7, 8, 9]\n",
    "子进程2 3.14\n",
    "子进程2 [10, 10, 2, 3, 4, 5, 6, 7, 8, 9]\n",
    "父进程 3.14\n",
    "父进程 [10, 10, 2, 3, 4, 5, 6, 7, 8, 9]\n",
    "No Except\n",
    "```\n",
    "类型方面的对应关系：\n",
    "```py\n",
    "typecode_to_type = {\n",
    "    'c': ctypes.c_char,\n",
    "    'u': ctypes.c_wchar,\n",
    "    'b': ctypes.c_byte,\n",
    "    'B': ctypes.c_ubyte,\n",
    "    'h': ctypes.c_short,\n",
    "    'H': ctypes.c_ushort,\n",
    "    'i': ctypes.c_int,\n",
    "    'I': ctypes.c_uint,\n",
    "    'l': ctypes.c_long,\n",
    "    'L': ctypes.c_ulong,\n",
    "    'q': ctypes.c_longlong,\n",
    "    'Q': ctypes.c_ulonglong,\n",
    "    'f': ctypes.c_float,\n",
    "    'd': ctypes.c_double\n",
    "}\n",
    "```\n",
    "这两个类型其实是`ctypes`类型，更多的类型可以去<a href=\"https://docs.python.org/3/library/multiprocessing.html#module-multiprocessing.sharedctypes\" target=\"_blank\">` multiprocessing.sharedctypes`</a>查看，来张图：\n",
    "![4.ctypes.png](https://images2018.cnblogs.com/blog/1127869/201808/1127869-20180815162731961-1455673124.png)\n",
    "回头解决`GIL`的时候会用到`C`系列或者`Go`系列的共享库（讲线程的时候会说）\n",
    "\n",
    "---\n",
    "\n",
    "关于进程安全的补充说明：对于原子性操作就不用说，铁定安全，但注意一下`i+=1`并不是原子性操作：\n",
    "```py\n",
    "from multiprocessing import Process, Value\n",
    "\n",
    "def proc_test1(value):\n",
    "    for i in range(1000):\n",
    "        value.value += 1\n",
    "\n",
    "def main():\n",
    "    value = Value(\"i\", 0)\n",
    "    p_list = [Process(target=proc_test1, args=(value, )) for i in range(5)]\n",
    "    # 批量启动\n",
    "    for i in p_list:\n",
    "        i.start()\n",
    "    # 批量资源回收\n",
    "    for i in p_list:\n",
    "        i.join()\n",
    "    print(value.value)\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    main()\n",
    "```\n",
    "输出：（理论上应该是:5×1000=5000）\n",
    "```\n",
    "2153\n",
    "```\n",
    "稍微改一下才行：（**进程安全：只是提供了安全的方法，并不是什么都不用你操心了**）\n",
    "```py\n",
    "# 通用方法\n",
    "def proc_test1(value):\n",
    "    for i in range(1000):\n",
    "        if value.acquire():\n",
    "            value.value += 1\n",
    "        value.release()\n",
    "\n",
    "# 官方案例：(Lock可以使用with托管)\n",
    "def proc_test1(value):\n",
    "    for i in range(1000):\n",
    "        with value.get_lock():\n",
    "            value.value += 1\n",
    "\n",
    "# 更多可以查看：`sharedctypes.SynchronizedBase` 源码\n",
    "```\n",
    "输出：（关于锁这块，后面讲线程的时候会详说，看看就好【语法的确比C#麻烦点】）\n",
    "```\n",
    "5000\n",
    "```\n",
    "看看源码：（之前探讨如何优雅的杀死子进程，其中就有一种方法使用了`Value`）\n",
    "```py\n",
    "def Value(typecode_or_type, *args, lock=True, ctx=None):\n",
    "    '''返回Value的同步包装器'''\n",
    "    obj = RawValue(typecode_or_type, *args)\n",
    "    if lock is False:\n",
    "        return obj\n",
    "    # 默认支持Lock\n",
    "    if lock in (True, None):\n",
    "        ctx = ctx or get_context() # 获取上下文\n",
    "        lock = ctx.RLock() # 获取递归锁\n",
    "    if not hasattr(lock, 'acquire'): \n",
    "        raise AttributeError(\"%r has no method 'acquire'\" % lock)\n",
    "    # 一系列处理\n",
    "    return synchronized(obj, lock, ctx=ctx)\n",
    "\n",
    "def Array(typecode_or_type, size_or_initializer, *, lock=True, ctx=None):\n",
    "    '''返回RawArray的同步包装器'''\n",
    "    obj = RawArray(typecode_or_type, size_or_initializer)\n",
    "    if lock is False:\n",
    "        return obj\n",
    "    # 默认是支持Lock的\n",
    "    if lock in (True, None):\n",
    "        ctx = ctx or get_context() # 获取上下文\n",
    "        lock = ctx.RLock()  # 递归锁属性\n",
    "    # 查看是否有acquire属性\n",
    "    if not hasattr(lock, 'acquire'):\n",
    "        raise AttributeError(\"%r has no method 'acquire'\" % lock)\n",
    "    return synchronized(obj, lock, ctx=ctx)\n",
    "```\n",
    "\n",
    "扩展部分可以查看这篇文章：<a href=\"http://blog.51cto.com/11026142/1874807\" target=\"_blank\">http://blog.51cto.com/11026142/1874807</a>\n",
    "\n",
    "---\n",
    "\n",
    "#### 2.服务器进程（`Manager`）\n",
    "\n",
    "官方文档：<a href=\"https://docs.python.org/3/library/multiprocessing.html#managers\" target=\"_blank\">https://docs.python.org/3/library/multiprocessing.html#managers</a>\n",
    "\n",
    "**有一个服务器进程负责维护所有的对象，而其他进程连接到该进程，通过代理对象操作服务器进程当中的对象**\n",
    "\n",
    "通过返回的经理`Manager()`将支持类型`list、dict、Namespace、Lock、RLock、Semaphore、BoundedSemaphore、Condition、Event、Barrier、Queue`\n",
    "\n",
    "举个简单例子（后面还会再说）：(本质其实就是`多个进程通过代理，共同操作服务端内容`)\n",
    "```py\n",
    "from multiprocessing import Pool, Manager\n",
    "\n",
    "def test1(d, l):\n",
    "    d[1] = '1'\n",
    "    d['2'] = 2\n",
    "    d[0.25] = None\n",
    "    l.reverse()\n",
    "\n",
    "def test2(d, l):\n",
    "    print(d)\n",
    "    print(l)\n",
    "\n",
    "def main():\n",
    "    with Manager() as manager:\n",
    "        dict_test = manager.dict()\n",
    "        list_test = manager.list(range(10))\n",
    "\n",
    "        pool = Pool()\n",
    "        pool.apply_async(test1, args=(dict_test, list_test))\n",
    "        pool.apply_async(test2, args=(dict_test, list_test))\n",
    "        pool.close()\n",
    "        pool.join()\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    main()\n",
    "```\n",
    "输出：\n",
    "```\n",
    "{1: '1', '2': 2, 0.25: None}\n",
    "[9, 8, 7, 6, 5, 4, 3, 2, 1, 0]\n",
    "```\n",
    "服务器进程管理器比使用共享内存对象更灵活，因为它们可以支持任意对象类型。此外，单个管理器可以通过网络在不同计算机上的进程共享。但是，它们比使用共享内存慢（毕竟有了`“中介”`）\n",
    "\n",
    "同步问题依然需要注意一下，举个例子体会一下：\n",
    "```py\n",
    "from multiprocessing import Manager, Process, Lock\n",
    "\n",
    "def test(dict1, lock):\n",
    "    for i in range(100):\n",
    "        with lock:  # 你可以把这句话注释掉，然后就知道为什么加了\n",
    "            dict1[\"year\"] += 1\n",
    "\n",
    "def main():\n",
    "    with Manager() as m:\n",
    "        lock = Lock()\n",
    "        dict1 = m.dict({\"year\": 2000})\n",
    "        p_list = [Process(target=test, args=(dict1, lock)) for i in range(5)]\n",
    "        for i in p_list:\n",
    "            i.start()\n",
    "        for i in p_list:\n",
    "            i.join()\n",
    "        print(dict1)\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    main()\n",
    "```\n",
    "\n",
    "扩展补充：\n",
    "1. `multiprocessing.Lock`是一个进程安全对象，因此您可以将其直接传递给子进程并在所有进程中安全地使用它。\n",
    "2. 大多数可变Python对象（如list，dict,大多数类）不能保证进程中安全，所以它们在进程间共享时需要使用`Manager`\n",
    "3. 多进程模式的缺点是创建进程的代价大，在`Unix/Linux`系统下，用`fork`调用还行，在`Windows`下创建进程开销巨大。\n",
    "\n",
    "Manager这块官方文档很详细，可以看看：<a href=\"https://docs.python.org/3/library/multiprocessing.html#managers\" target=\"_blank\">https://docs.python.org/3/library/multiprocessing.html#managers</a>\n",
    "\n",
    "\n",
    "`WinServer`的可以参考<a href=\"https://www.cnblogs.com/bin-l/p/8615201.html\" target=\"_blank\">这篇</a> or <a href=\"https://blog.csdn.net/Solo95/article/details/78913709\" target=\"_blank\">这篇埋坑记</a>（Manager一般都是部署在Linux的，Win的客户端不影响）\n",
    "\n",
    "#### 扩展补充\n",
    "\n",
    "还记得之前的：[**无法将multiprocessing.Queue对象传递给Pool方法**](#进程间通信2)吗？其实一般都是这两种方式解决的：\n",
    "1. 使用Manager需要生成另一个进程来托管Manager服务器。 并且所有获取/释放锁的调用都必须通过IPC发送到该服务器。\n",
    "2. 使用初始化程序在池创建时传递常规`multiprocessing.Queue()`这将使`Queue`实例在所有子进程中全局共享\n",
    "\n",
    "再看一下Pool的`__init__`方法：\n",
    "```py\n",
    "# processes：进程数\n",
    "# initializer,initargs 初始化进行的操作\n",
    "# maxtaskperchild：每个进程执行task的最大数目\n",
    "# contex：上下文对象\n",
    "def __init__(self, processes=None, initializer=None, initargs=(),\n",
    "                 maxtasksperchild=None, context=None):\n",
    "```\n",
    "第一种方法不够轻量级，在讲案例前，稍微说下第二种方法：(也算把上面留下的悬念解了)\n",
    "```py\n",
    "import os\n",
    "import time\n",
    "from multiprocessing import Pool, Queue\n",
    "\n",
    "def error_callback(msg):\n",
    "    print(msg)\n",
    "\n",
    "def pro_test1():\n",
    "    print(\"[子进程1]PPID=%d,PID=%d\" % (os.getppid(), os.getpid()))\n",
    "    q.put(\"[子进程1]小明，今晚撸串不？\")\n",
    "\n",
    "    # 设置一个简版的重试机制（三次重试）\n",
    "    for i in range(3):\n",
    "        if not q.empty():\n",
    "            print(q.get())\n",
    "            break\n",
    "        else:\n",
    "            time.sleep((i + 1) * 2)  # 第一次1s，第二次4s，第三次6s\n",
    "\n",
    "def pro_test2():\n",
    "    print(\"[子进程2]PPID=%d,PID=%d\" % (os.getppid(), os.getpid()))\n",
    "    print(q.get())\n",
    "    time.sleep(4)  # 模拟一下网络延迟\n",
    "    q.put(\"[子进程2]不去，我今天约了妹子\")\n",
    "\n",
    "def init(queue):\n",
    "    global q\n",
    "    q = queue\n",
    "\n",
    "def main():\n",
    "    print(\"[父进程]PPID=%d,PID=%d\" % (os.getppid(), os.getpid()))\n",
    "    queue = Queue()\n",
    "    p = Pool(initializer=init, initargs=(queue, ))\n",
    "    p.apply_async(pro_test1, error_callback=error_callback)\n",
    "    p.apply_async(pro_test2, error_callback=error_callback)\n",
    "    p.close()\n",
    "    p.join()\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    main()\n",
    "```\n",
    "输出：（就是在初始化Pool的时候，**传了初始化执行的方法并传了参数**：`alizer=init, initargs=(queue, ))`）\n",
    "```\n",
    "[父进程]PPID=13157,PID=24864\n",
    "[子进程1]PPID=24864,PID=24865\n",
    "[子进程2]PPID=24864,PID=24866\n",
    "[子进程1]小明，今晚撸串不？\n",
    "[子进程2]不去，我今天约了妹子\n",
    "\n",
    "real    0m6.105s\n",
    "user    0m0.071s\n",
    "sys     0m0.042s\n",
    "```\n",
    "Win下亦通用（win下没有`os.getgid`）\n",
    "![5.win.png](https://images2018.cnblogs.com/blog/1127869/201808/1127869-20180816163829040-2076694825.png)\n",
    "\n",
    "---\n",
    "\n",
    "### 1.7.分布式进程的案例\n",
    "\n",
    "有了`1.6`的基础，咱们来个例子练练：\n",
    "\n",
    "`BaseManager`的缩略图：\n",
    "![6.缩略.png](https://images2018.cnblogs.com/blog/1127869/201808/1127869-20180816165142638-798135918.png)\n",
    "\n",
    "服务器端代码：\n",
    "```py\n",
    "from multiprocessing import Queue\n",
    "from multiprocessing.managers import BaseManager\n",
    "\n",
    "def main():\n",
    "    # 用来身份验证的\n",
    "    key = b\"8d969eef6ecad3c29a3a629280e686cf0c3f5d5a86aff3ca12020c923adc6c92\"\n",
    "    get_zhang_queue = Queue()  # 小张消息队列\n",
    "    get_ming_queue = Queue()  # 小明消息队列\n",
    "\n",
    "    # 把Queue注册到网络上, callable参数关联了Queue对象\n",
    "    BaseManager.register(\"get_zhang_queue\", callable=lambda: get_zhang_queue)\n",
    "    BaseManager.register(\"get_ming_queue\", callable=lambda: get_ming_queue)\n",
    "\n",
    "    # 实例化一个Manager对象。绑定ip+端口, 设置验证秘钥\n",
    "    manager = BaseManager(address=(\"192.168.36.235\", 5438), authkey=key)\n",
    "    # 运行serve\n",
    "    manager.get_server().serve_forever()\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    main()\n",
    "```\n",
    "客户端代码1：\n",
    "```py\n",
    "from multiprocessing.managers import BaseManager\n",
    "\n",
    "def main():\n",
    "    \"\"\"客户端1\"\"\"\n",
    "    key = b\"8d969eef6ecad3c29a3a629280e686cf0c3f5d5a86aff3ca12020c923adc6c92\"\n",
    "\n",
    "    # 注册对应方法的名字（从网络上获取Queue）\n",
    "    BaseManager.register(\"get_ming_queue\")\n",
    "    BaseManager.register(\"get_zhang_queue\")\n",
    "\n",
    "    # 实例化一个Manager对象。绑定ip+端口, 设置验证秘钥\n",
    "    m = BaseManager(address=(\"192.168.36.235\", 5438), authkey=key)\n",
    "    # 连接到服务器\n",
    "    m.connect()\n",
    "\n",
    "    q1 = m.get_zhang_queue()  # 在自己队列里面留言\n",
    "    q1.put(\"[小张]小明，老大明天是不是去外地办事啊？\")\n",
    "\n",
    "    q2 = m.get_ming_queue()  # 获取小明说的话\n",
    "    print(q2.get())\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    main()\n",
    "```\n",
    "客户端代码2：\n",
    "```\n",
    "from multiprocessing.managers import BaseManager\n",
    "\n",
    "def main():\n",
    "    \"\"\"客户端2\"\"\"\n",
    "    key = b\"8d969eef6ecad3c29a3a629280e686cf0c3f5d5a86aff3ca12020c923adc6c92\"\n",
    "    \n",
    "    # 注册对应方法的名字（从网络上获取Queue）\n",
    "    BaseManager.register(\"get_ming_queue\")\n",
    "    BaseManager.register(\"get_zhang_queue\")\n",
    "\n",
    "    # 实例化一个Manager对象。绑定ip+端口, 设置验证秘钥\n",
    "    m = BaseManager(address=(\"192.168.36.235\", 5438), authkey=key)\n",
    "    # 连接到服务器\n",
    "    m.connect()\n",
    "\n",
    "    q1 = m.get_zhang_queue()  # 获取小张说的话\n",
    "    print(q1.get())\n",
    "\n",
    "    q2 = m.get_ming_queue()  # 在自己队列里面留言\n",
    "    q2.put(\"[小明]这几天咱们终于可以不加班了(>_<)\")\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    main()\n",
    "```\n",
    "输出图示：\n",
    "![7.manager.gif](https://images2018.cnblogs.com/blog/1127869/201808/1127869-20180816180721761-490336831.gif)\n",
    "服务器运行在Linux的测试：\n",
    "![8.win.png](https://images2018.cnblogs.com/blog/1127869/201808/1127869-20180816181817723-1399170499.png)\n",
    "\n",
    "其实还有一部分内容没说，明天得出去办点事，先到这吧，后面找机会继续带一下\n",
    "\n",
    "---\n",
    "\n",
    "参考文章：\n",
    "\n",
    "进程共享的探讨：<a href=\"https://stackoverflow.com/questions/25557686/python-sharing-a-lock-between-processes\" target=\"_blank\">python-sharing-a-lock-between-processes</a>\n",
    "\n",
    "多进程锁的探讨：<a href=\"https://stackoverflow.com/questions/17960296/trouble-using-a-lock-with-multiprocessing-pool-pickling-error\" target=\"_blank\">trouble-using-a-lock-with-multiprocessing-pool-pickling-error</a>\n",
    "\n",
    "JoinableQueue扩展：<a href=\"https://www.cnblogs.com/smallmars/p/7093603.html\" target=\"_blank\">https://www.cnblogs.com/smallmars/p/7093603.html</a>\n",
    "\n",
    "Python多进程编程：<a href=\"https://www.cnblogs.com/kaituorensheng/p/4445418.html\" target=\"_blank\">https://www.cnblogs.com/kaituorensheng/p/4445418.html</a>\n",
    "\n",
    "有深度但需要辩证看的两篇文章：\n",
    "\n",
    "跨进程对象共享：<a href=\"http://blog.ftofficer.com/2009/12/python-multiprocessing-3-about-queue/\" target=\"_blank\">http://blog.ftofficer.com/2009/12/python-multiprocessing-3-about-queue</a>\n",
    "\n",
    "关于Queue：<a href=\"http://blog.ftofficer.com/2009/12/python-multiprocessing-2-object-sharing-across-process\" target=\"_blank\">http://blog.ftofficer.com/2009/12/python-multiprocessing-2-object-sharing-across-process</a>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## NetCore并发编程\n",
    "\n",
    "示例代码：<a href=\"https://github.com/lotapp/BaseCode/tree/master/netcore/4_Concurrency\" target=\"_blank\">https://github.com/lotapp/BaseCode/tree/master/netcore/4_Concurrency</a>\n",
    "\n",
    "先简单说下概念（其实之前也有说，所以简说下）：\n",
    "1. 并发：同时做多件事情\n",
    "2. 多线程：并发的一种形式\n",
    "3. 并行处理：多线程的一种（线程池产生的一种并发类型，eg：**异步编程**）\n",
    "4. 响应式编程：一种编程模式，对事件进行响应（有点类似于JQ的事件）\n",
    "\n",
    "Net里面很少用进程，在以前基本上都是`线程+池+异步+并行+协程`\n",
    "\n",
    "我这边简单引入一下，毕竟主要是写Python的教程，Net只是帮你们回顾一下，如果你发现还没听过这些概念，或者你的项目中还充斥着各种`Thread`和`ThreadPool`的话，真的得系统的学习一下了，现在官网的文档已经很完善了，记得早几年啥都没有，也只能挖那些外国开源项目：\n",
    "\n",
    "<a href=\"https://docs.microsoft.com/zh-cn/dotnet/standard/parallel-processing-and-concurrency\" target=\"_blank\">https://docs.microsoft.com/zh-cn/dotnet/standard/parallel-processing-and-concurrency</a>\n",
    "\n",
    "### 1.异步编程（Task）\n",
    "\n",
    "Task的目的其实就是为了简化`Thread`和`ThreadPool`的代码，下面一起看看吧：\n",
    "\n",
    "异步用起来比较简单，一般IO，DB，Net用的比较多，很多时候都会采用重试机制，举个简单的例子：\n",
    "```csharp\n",
    "/// <summary>\n",
    "/// 模拟一个网络操作（别忘了重试机制）\n",
    "/// </summary>\n",
    "/// <param name=\"url\">url</param>\n",
    "/// <returns></returns>\n",
    "private async static Task<string> DownloadStringAsync(string url)\n",
    "{\n",
    "    using (var client = new HttpClient())\n",
    "    {\n",
    "        // 设置第一次重试时间\n",
    "        var nextDelay = TimeSpan.FromSeconds(1);\n",
    "        for (int i = 0; i < 3; i++)\n",
    "        {\n",
    "            try\n",
    "            {\n",
    "                return await client.GetStringAsync(url);\n",
    "            }\n",
    "            catch { }\n",
    "            await Task.Delay(nextDelay); // 用异步阻塞的方式防止服务器被太多重试给阻塞了\n",
    "            nextDelay *= 2; // 3次重试机会，第一次1s，第二次2s，第三次4s\n",
    "        }\n",
    "        // 最后一次尝试，错误就抛出\n",
    "        return await client.GetStringAsync(url);\n",
    "    }\n",
    "}\n",
    "```\n",
    "然后补充说下Task异常的问题，当你await的时候如果有异常会抛出，在第一个await处捕获处理即可\n",
    "\n",
    "如果`async`和`await`就是理解不了的可以这样想：`async`就是为了让`await`生效（为了向后兼容）\n",
    "\n",
    "对了，如果返回的是void，你设置成Task就行了，触发是类似于事件之类的方法才使用void，不然没有返回值都是使用Task\n",
    "\n",
    "项目里经常有这么一个场景：**等待一组任务完成后再执行某个操作**,看个引入案例：\n",
    "```csharp\n",
    "/// <summary>\n",
    "/// 1.批量任务\n",
    "/// </summary>\n",
    "/// <param name=\"list\"></param>\n",
    "/// <returns></returns>\n",
    "private async static Task<string[]> DownloadStringAsync(IEnumerable<string> list)\n",
    "{\n",
    "    using (var client = new HttpClient())\n",
    "    {\n",
    "        var tasks = list.Select(url => client.GetStringAsync(url)).ToArray();\n",
    "        return await Task.WhenAll(tasks);\n",
    "    }\n",
    "}\n",
    "```\n",
    "再举一个场景：**同时调用多个同效果的API，有一个返回就好了，其他的忽略**\n",
    "```csharp\n",
    "/// <summary>\n",
    "/// 2.返回首先完成的Task\n",
    "/// </summary>\n",
    "/// <param name=\"list\"></param>\n",
    "/// <returns></returns>\n",
    "private static async Task<string> GetIPAsync(IEnumerable<string> list)\n",
    "{\n",
    "    using (var client = new HttpClient())\n",
    "    {\n",
    "        var tasks = list.Select(url => client.GetStringAsync(url)).ToArray();\n",
    "        var task = await Task.WhenAny(tasks); // 返回第一个完成的Task\n",
    "        return await task;\n",
    "    }\n",
    "}\n",
    "```\n",
    "一个async方法被await调用后，当它恢复运行时就会回到原来的上下文中运行。\n",
    "\n",
    "如果你的Task不再需要上下文了可以使用：`task.ConfigureAwait(false)`，eg：写个日记还要啥上下文？\n",
    "\n",
    "逆天的建议是：**在核心代码里面一种使用`ConfigureAwait`，用户页面相关代码，不需要上下文的加上**\n",
    "\n",
    "其实如果有太多await在上下文里恢复那也是比较卡的，使用`ConfigureAwait`之后，被暂停后会在线程池里面继续运行\n",
    "\n",
    "再看一个场景：比如一个耗时操作，我需要指定它的超时时间：\n",
    "```csharp\n",
    " /// <summary>\n",
    "/// 3.超时取消\n",
    "/// </summary>\n",
    "/// <returns></returns>\n",
    "private static async Task<string> CancellMethod()\n",
    "{\n",
    "    //实例化取消任务\n",
    "    var cts = new CancellationTokenSource();\n",
    "    cts.CancelAfter(TimeSpan.FromSeconds(3)); // 设置失效时间为3s\n",
    "    try\n",
    "    {\n",
    "        return await DoSomethingAsync(cts.Token);\n",
    "    }\n",
    "    // 任务已经取消会引发TaskCanceledException\n",
    "    catch (TaskCanceledException ex)\n",
    "    {\n",
    "\n",
    "        return \"false\";\n",
    "    }\n",
    "}\n",
    "/// <summary>\n",
    "/// 模仿一个耗时操作\n",
    "/// </summary>\n",
    "/// <returns></returns>\n",
    "private static async Task<string> DoSomethingAsync(CancellationToken token)\n",
    "{\n",
    "    await Task.Delay(TimeSpan.FromSeconds(5), token);\n",
    "    return \"ok\";\n",
    "}\n",
    "```\n",
    "异步这块简单回顾就不说了，留两个扩展，你们自行探讨：\n",
    "1. 进度方面的可以使用`IProgress<T>`，就当留个作业自己摸索下吧～\n",
    "2. 使用了异步之后尽量避免使用`task.Wait` or `task.Result`，这样可以避免死锁\n",
    "\n",
    "Task其他新特征去官网看看吧，引入到此为止了。\n",
    "\n",
    "---\n",
    "\n",
    "### 2.并行编程（Parallel）\n",
    "\n",
    "这个其实出来很久了，现在基本上都是用`PLinq`比较多点，主要就是：\n",
    "1. **数据并行**：重点在处理数据（eg：聚合）\n",
    "2. **任务并行**：重点在执行任务（每个任务块尽可能独立，越独立效率越高）\n",
    "\n",
    "#### 数据并行\n",
    "\n",
    "以前都是`Parallel.ForEach`这么用，现在和Linq结合之后非常方便`.AsParallel()`就OK了\n",
    "\n",
    "说很抽象看个简单案例：\n",
    "\n",
    "```csharp\n",
    "static void Main(string[] args)\n",
    "{\n",
    "    IEnumerable<int> list = new List<int>() { 1, 2, 3, 4, 5, 7, 8, 9 };\n",
    "    foreach (var item in ParallelMethod(list))\n",
    "    {\n",
    "        Console.WriteLine(item);\n",
    "    }\n",
    "}\n",
    "/// <summary>\n",
    "/// 举个例子\n",
    "/// </summary>\n",
    "private static IEnumerable<int> ParallelMethod(IEnumerable<int> list)\n",
    "{\n",
    "    return list.AsParallel().Select(x => x * x);\n",
    "}\n",
    "```\n",
    "正常执行的结果应该是：\n",
    "```\n",
    "1\n",
    "4\n",
    "9\n",
    "25\n",
    "64\n",
    "16\n",
    "49\n",
    "81\n",
    "```\n",
    "并行之后就是这样了（不管顺序了）：\n",
    "```\n",
    "25\n",
    "64\n",
    "1\n",
    "9\n",
    "49\n",
    "81\n",
    "4\n",
    "16\n",
    "```\n",
    "\n",
    "当然了，如果你就是对顺序有要求可以使用：**`.AsOrdered()`**\n",
    "```csharp\n",
    "/// <summary>\n",
    "/// 举个例子\n",
    "/// </summary>\n",
    "private static IEnumerable<int> ParallelMethod(IEnumerable<int> list)\n",
    "{\n",
    "    return list.AsParallel().AsOrdered().Select(x => x * x);\n",
    "}\n",
    "```\n",
    "\n",
    "其实实际项目中，使用并行的时候：**任务时间适中，太长不适合，太短也不适合**\n",
    "\n",
    "记得大家在项目里经常会用到如`Sum`，`Count`等聚合函数，其实这时候使用并行就很合适\n",
    "\n",
    "```csharp\n",
    "var list = new List<long>();\n",
    "for (long i = 0; i < 1000000; i++)\n",
    "{\n",
    "    list.Add(i);\n",
    "}\n",
    "Console.WriteLine(GetSumParallel(list));\n",
    "```\n",
    "```csharp\n",
    "private static long GetSumParallel(IEnumerable<long> list)\n",
    "{\n",
    "    return list.AsParallel().Sum();\n",
    "}\n",
    "```\n",
    "time dotnet PLINQ.dll\n",
    "```\n",
    "499999500000\n",
    "\n",
    "real\t0m0.096s\n",
    "user\t0m0.081s\n",
    "sys\t0m0.025s\n",
    "```\n",
    "不使用并行：（稍微多了点，CPU越密集差距越大）\n",
    "```\n",
    "499999500000\n",
    "\n",
    "real\t0m0.103s\n",
    "user\t0m0.092s\n",
    "sys\t0m0.021s\n",
    "```\n",
    "其实聚合有一个通用方法，可以支持复杂的聚合：(以上面sum为例)\n",
    "```\n",
    ".Aggregate(\n",
    "            seed:0,\n",
    "            func:(sum,item)=>sum+item\n",
    "          );\n",
    "```\n",
    "\n",
    "稍微扩展一下，PLinq也是支持取消的，**`.WithCancellation(CancellationToken)`**\n",
    "\n",
    "Token的用法和上面一样，就不复述了，如果需要和异步结合，一个`Task.Run`就可以把并行任务交给线程池了\n",
    "\n",
    "也可以使用Task的异步方法，设置超时时间，这样PLinq超时了也就终止了\n",
    "\n",
    "PLinq这么方便，其实也是有一些小弊端的，比如它会直接最大程度的占用系统资源，可能会影响其他的任务，而传统的Parallel则会动态调整\n",
    "\n",
    "---\n",
    "\n",
    "#### 任务并行（并行调用）\n",
    "\n",
    "这个PLinq好像没有对应的方法，有新语法你可以说下，来举个例子：\n",
    "```csharp\n",
    "await Task.Run(() =>\n",
    "    Parallel.Invoke(\n",
    "        () => Task.Delay(TimeSpan.FromSeconds(3)),\n",
    "        () => Task.Delay(TimeSpan.FromSeconds(2))\n",
    "    ));\n",
    "```\n",
    "取消也支持：\n",
    "```csharp\n",
    "Parallel.Invoke(new ParallelOptions() { CancellationToken = token }, actions);\n",
    "```\n",
    "\n",
    "### 扩充说明\n",
    "\n",
    "其实还有一些比如**数据流**和**响应编程**没说，这个之前都是用第三方库，刚才看官网文档，好像已经支持了，所以就不卖弄了，感兴趣的可以去看看，其实项目里面有流数据相关的框架，eg：`Spark`，都是比较成熟的解决方案了基本上也不太使用这些了。\n",
    "\n",
    "然后还有一些没说，比如NetCore里面**不可变类型**（列表、字典、集合、队列、栈、线程安全字典等等）以及**限流**、**任务调度**等，这些关键词我提一下，也方便你去搜索自己学习拓展\n",
    "\n",
    "先到这吧，其他的自己探索一下吧，最后贴一些Nuget库，你可以针对性的使用：\n",
    "\n",
    "1. **数据流**：`Microsoft.Tpl.Dataflow`\n",
    "2. **响应编程**(Linq的Rx操作)：`Rx-Main`\n",
    "3. **不可变类型**：`Microsoft.Bcl.Immutable`\n",
    "\n",
    "不得不感慨一句，微软妈妈真的花了很多功夫，Net的并发编程比Python省心多了（完）"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
